/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 下午3:37:02
*/
package cn.com.guage.flink.sink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @version:1.0.1
 * @Description: 数据写入到Redis 自定义sink
 * @author: yangdechao
 * @date: datedate 2021年11月15日 下午3:37:02
 */
public class JedisSinkTest {
	/**
	 * 使用自定义分区 根据数字的奇偶性来进行分区
	 *
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		DataStreamSource<String> textStream = env.readTextFile("data/redis-data.txt");
		textStream.filter(t -> t.split(",").length > 2).map(new MapFunction<String, Tuple3<String, String, Long>>() {
			/**
			 * 
			 */
			private static final long serialVersionUID = -4668565534478998301L;

			@Override
			public Tuple3<String, String, Long> map(String value) throws Exception {
				return new Tuple3<String, String, Long>(value.split(",")[0], value.split(",")[1],
						Long.parseLong(value.split(",")[2]));
			}
		}).addSink(new GuageRedisSink());
		env.execute("Test");

	}
}
